#pragma once

#include "AbstractRequestService.h"
#include "model/ServerNode.h"
#include "base/TokenBucket.h"
#include "base/ThreadPool.hpp"

namespace zeroplus::mq
{
    class OutsideService : public AbstractRequestService
    {
    private:
        std::unique_ptr<ServerNode> server_; // 处理生产者、消费者的心跳，返回ack给生产者、接收消费者的ack
        std::unordered_map<transport::RequestCode, std::shared_ptr<AbstractRequestService>> requestTable_;
        TokenBucket<std::chrono::steady_clock> tokenBucket_;
        ThreadPool threadPool_;

    public:
        OutsideService(std::shared_ptr<BrokerMediator> mediator)
            : AbstractRequestService(mediator),
              tokenBucket_(100, 1000),
              threadPool_(8)
        {
            server_ = std::make_unique<ServerNode>(BrokerConfig::getInstance().serverAddr);
        }

        ~OutsideService()
        {
            threadPool_.Shutdown(); // 关闭线程池
        }

        void OutSideServicehandler()
        {
            while (true)
            {
                //  LOG_DEBUG("处理客户端请求");
                Bytes requestCmdBytes = server_->RecvRequest();
                // 检查令牌桶是否有可用的令牌
                if (!tokenBucket_.consume(1))
                {
                    SendRateLimitResponse();
                    return; // 没有可用的令牌，拒绝本次请求
                }

                // 队列满载保护
                if (threadPool_.IsFull())
                {
                    SendBusyResponse();
                    return;
                }

                // 提交任务
                auto task = [this, requestCmdBytes]()
                {
                    auto requestCmd = ProtocolCodec<transport::RequestCommand>::Decode(requestCmdBytes);
                    auto replyCmd = ProcessRequest(requestCmd);
                    if (replyCmd.has_value())
                    {
                        server_->SendResponse(ProtocolCodec<transport::ReplyCommand>::Encode(replyCmd.value()));
                    }
                };
                // task();
                if (!threadPool_.PostTask(task))
                {
                    SendBusyResponse();
                }
            }
        }
        void AddRequestCode(const transport::RequestCode &type, std::shared_ptr<AbstractRequestService> requestService)
        {
            requestTable_[type] = requestService;
        }

        zmqpp::socket_t *GetSocket()
        {
            return server_->GetSocket();
        }

        std::optional<transport::ReplyCommand> ProcessRequest(const transport::RequestCommand &requestCmd)
        {
            transport::RequestCode requestCode = requestCmd.code();
            if (requestTable_.find(requestCode) == requestTable_.end())
            {
                LOG_FATAL("No command registered for request code: %d", static_cast<int>(requestCode));
            }
            return requestTable_[requestCode]->ProcessRequest(requestCmd);
        }

        void SendRateLimitResponse()
        {
            LOG_WARN("请求速率过快，已限流");
            transport::ReplyCommand reply;
            reply.set_code(transport::ResponseCode::HIGH_REQUEST_RATE);
            server_->SendResponse(ProtocolCodec<transport::ReplyCommand>::Encode(reply));
        }

        void SendBusyResponse()
        {
            transport::ReplyCommand reply;
            reply.set_code(transport::ResponseCode::SERVER_ERROR);
            reply.set_body("System busy, please retry later");
            server_->SendResponse(ProtocolCodec<transport::ReplyCommand>::Encode(reply));
        }
    };
}
